package com.feidee.fd.sml.algorithm.ml

import com.feidee.fd.sml.algorithm.component.ml.classification.{MultilayePerceptronComponent, MultilayerPerceptronParam}
import com.feidee.fd.sml.algorithm.util.{TestingDataGenerator, ToolClass}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.sql.types.DoubleType
import org.scalatest.FunSuite
/**
  * @author YongChen
  * @date 2018/12/25 10:18
  * @description 算法规则函数
  * @reviewer dongguosheng
  */
class MultilayerPerceptronSuite extends FunSuite {
  val mpc = new MultilayePerceptronComponent
  val paramStr: String =
    """
      |{
      |	'input_pt': 'b',
      |	'output_pt': 'a',
      | 'hive_table': '',
      | 'flow_time': '',
      |	'featuresCol': 'features',
      |	'labelCol': 'label',
      |	'modelPath': '',
      |	'metrics': ['accuracy', 'auc', 'pr', 'abc'] ,
      | 'layers': ['4', '5', '4', '3'] ,
      | 'blockSize': 128,
      | 'solver' : 'l-bfgs' ,
      | 'maxIter' : 100 ,
      | 'tol' : 1E-6 ,
      | 'seed' : 1234,
      | 'stepSize' : 0.03
      |}
    """.stripMargin
  val param: MultilayerPerceptronParam = mpc.parseParam(new ToolClass().encrypt(paramStr))
  print(param)

  /**
    * MultilayerPerceptron 参数构造方法测试：检测到读取出合法参数 & 成功赋值默认参数，即认为测试通过
    */
  test("MultilayerPerceptron parameter construction test") {
    assert(
      "b".equals(param.input_pt)
        & "a".equals(param.output_pt)
        & "".equals(param.hive_table)
        & "".equals(param.flow_time)
        & "features".equals(param.featuresCol)
        & "label".equals(param.labelCol)
        & "".equals(param.modelPath)
        & param.metrics.sameElements(Array("accuracy", "auc", "pr", "abc"))
        & (param.layers sameElements Array(4, 5, 4, 3))
        & param.blockSize == 128
        & "l-bfgs".equals(param.solver)
        & param.maxIter == 100
        & param.tol == 1E-6
        & param.seed == 1234
        & param.stepSize == 0.03
    )
  }


  /**
    * GBDT 模型训练功能测试：传入生成的测试数据和正确参数，可以正常训练 MultilayerPerceptron 模型，且预测结果列等于生成数据列数，即认为测试通过
    */
  test("MultilayerPerceptron training function test") {

    param.verify()
    // 测试模型训练方法
    val testingData = TestingDataGenerator.sampleMulticlassClassificationData


    val splits = testingData.randomSplit(Array(0.6, 0.4), seed = 1234L)
    val train = splits(0)
    val test = splits(1)

    val model = mpc.train(param, train)

    val result = model.transform(test)
    import TestingDataGenerator.spark.implicits._
    val predictionAndLabels = result.select("prediction", "label").withColumn("prediction", $"prediction".cast(DoubleType))
    predictionAndLabels.show(truncate = false)
    val evaluator = new MulticlassClassificationEvaluator().setPredictionCol("prediction").setLabelCol("label").setMetricName("accuracy")

    println("Test set accuracy = " + evaluator.evaluate(predictionAndLabels))

  }

}
